#pragma once

#include "Common.h"
#include "Base.h"
#include "Message.h"
#include "muduo/net/Buffer.h"
#include "muduo/net/TcpServer.h"
#include "muduo/net/TcpClient.h"
#include "muduo/net/TcpConnection.h"
#include "muduo/net/EventLoop.h"
#include "muduo/net/EventLoopThread.h"
#include "muduo/base/CountDownLatch.h"
#include <arpa/inet.h>
#include <mutex>
#include <unordered_map>

using namespace std::placeholders;

namespace rpc
{

    class MuduoBuffer : public BaseBuffer
    {
    public:
        using ptr = std::shared_ptr<MuduoBuffer>;

        MuduoBuffer(muduo::net::Buffer *buf)
            : _buf(buf)
        {
        }

        size_t readableSize() override
        {
            return _buf->readableBytes();
        }

        int32_t peekInt32() override
        {
            return _buf->peekInt32();
        }

        void retrieveInt32() override
        {
            _buf->retrieveInt32();
        }

        int32_t readInt32() override
        {
            return _buf->readInt32();
        }

        string retrieveAsString(size_t len) override
        {
            return _buf->retrieveAsString(len);
        }

    private:
        muduo::net::Buffer *_buf;
    };

    class BufferFactory
    {
    public:
        // 使用完美转发构建 MuduoBuffer 对象
        template <typename... Args>
        static BaseBuffer::ptr create(Args... args)
        {
            return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...);
        }
    };

    class LVProtocol : public BaseProtocol
    {
    public:
        using ptr = std::shared_ptr<LVProtocol>;
        // |--Len--|--VALUE--|
        // |--Len--|--mtype--|--idlen--|--id--|--body--|
        // 判断缓冲区中的数据量是否足够一条消息的处理
        bool canProcessed(const BaseBuffer::ptr &buf) override
        {
            if (buf->readableSize() < lenFieldLength)
            {
                DLOG("Net.h LVProtocol canProcessed() 1 buf->readableSize() = %d", buf->readableSize());
                return false;
            }

            int32_t total = buf->peekInt32();
            if (buf->readableSize() < (total + lenFieldLength))
            {
                DLOG("Net.h LVProtocol canProcessed() 2 total = %d, buf->readableSize() = %d", total, buf->readableSize());
                return false;
            }

            return true;
        }

        // 调用当前方法时默认缓冲区内有一条完整的消息待处理
        bool onMessage(const BaseBuffer::ptr &buf, BaseMessage::ptr &msg) override
        {
            int total = buf->readInt32();          // 读取总长度
            MType mtype = (MType)buf->readInt32(); // 读取mtype
            int idLen = buf->readInt32();          // 读取id的长度

            int bodyLen = total - mtypeFieldLength - idlenFieldLength - idLen;
            string id = buf->retrieveAsString(idLen); // 读取id
            string body = buf->retrieveAsString(bodyLen);

            // 使用消息工厂创建消息
            msg = MessageFactory::create(mtype);

            if (msg.get() == nullptr)
            {
                ELOG("消息创建失败!")
                return false;
            }

            bool ret = msg->unserialize(body);
            if (!ret)
            {
                ELOG("消息体反序列化失败")
                return false;
            }

            msg->setRid(id);
            msg->setMtype(mtype);

            return true;
        }

        // |--Len--|--mtype--|--idlen--|--id--|--body--|
        string serialize(BaseMessage::ptr &msg) override
        {
            DLOG("Net.h LVProtocol serialize() begin...");
            Json::Value val;
            string body = msg->serialize();
            string id = msg->rid();
            auto mtype = htonl((int32_t)msg->mtype());
            int32_t idlen = htonl(id.size());
            int32_t hTotalLen = mtypeFieldLength + idlenFieldLength + id.size() + body.size();
            int32_t nTotalLen = htonl(hTotalLen);

            DLOG("Net.h LVProtocol serialize() middle...");

            string result;
            result.reserve(hTotalLen);
            result.append((char *)&nTotalLen, lenFieldLength);
            result.append((char *)&mtype, mtypeFieldLength);
            result.append((char *)&idlen, idlenFieldLength);
            result.append(id);
            result.append(body);

            DLOG("Net.h LVProtocol serialize() hToTalLen = %d, len = %d",
                 hTotalLen, result.size());
            return result;
        }

    private:
        const size_t lenFieldLength = 4;
        const size_t mtypeFieldLength = 4;
        const size_t idlenFieldLength = 4;
    };

    class ProtocolFactory
    {
    public:
        template <typename... Args>
        static BaseProtocol::ptr create(Args... args)
        {
            return std::make_shared<LVProtocol>(std::forward<Args>(args)...);
        }
    };

    class MuduoConnection : public BaseConnection
    {
    public:
        using ptr = std::shared_ptr<MuduoConnection>;

        MuduoConnection(const muduo::net::TcpConnectionPtr &conn,
                        const BaseProtocol::ptr &protocol)
            : _conn(conn), _protocol(protocol) {}

        virtual void send(BaseMessage::ptr msg) override
        {
            string body = _protocol->serialize(msg);
            DLOG("Net.h MuduoConnection send() begin...");

            if (_conn->connected())
            {
                DLOG("Net.h MuduoConnection send() body = %s", body.c_str());
                _conn->send(body);
            }

            DLOG("Net.h MuduoConnection send() end...");
        }

        virtual void shutdown() override
        {
            _conn->shutdown();
        }

        virtual bool connected() override
        {
            return _conn->connected();
        }

    private:
        muduo::net::TcpConnectionPtr _conn;
        BaseProtocol::ptr _protocol;
    };

    class ConnectionFactory
    {
    public:
        template <typename... Args>
        static BaseConnection::ptr create(Args... args)
        {
            return std::make_shared<MuduoConnection>(std::forward<Args>(args)...);
        }
    };

    class MuduoService : public BaseService
    {
    public:
        using ptr = std::shared_ptr<MuduoService>;

        MuduoService(int port)
            : _server(&_loop, muduo::net::InetAddress("0.0.0.0", port), "MuduoService", muduo::net::TcpServer::kReusePort), _protocol(ProtocolFactory::create()) {}

        virtual void start()
        {
            _server.setConnectionCallback(std::bind(&MuduoService::onConnection, this, _1));
            _server.setMessageCallback(std::bind(&MuduoService::onMessage, this, _1, _2, _3));

            // 开始监听
            _server.start();
            DLOG("服务启动成功，开始监听事件...");
            // 死循环监控事件
            _loop.loop();
        }

    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                string info = conn->peerAddress().toIpPort() + " -> " + conn->localAddress().toIpPort() + " connect!";

                BaseConnection::ptr baseConn;
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    baseConn = ConnectionFactory::create(conn, _protocol);

                    _conns.insert(std::make_pair(conn, baseConn));
                }

                cout << info.c_str() << endl;
                // Todo: 回调函数
                if (_connectCb)
                    _connectCb(baseConn);
            }
            else
            {
                ILOG("...连接断开!")
                BaseConnection::ptr eraseConn;
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    auto it = _conns.find(conn);
                    if (it == _conns.end())
                    {
                        return;
                    }
                    eraseConn = it->second;
                    _conns.erase(conn);
                }

                // Todo: 回调函数
                if (_closeCb)
                    _closeCb(eraseConn);
            }
        }

        void onMessage(const muduo::net::TcpConnectionPtr &conn,
                       muduo::net::Buffer *buf,
                       muduo::Timestamp time)
        {
            cout << "连接有数据到来,开始处理!" << endl;
            BaseBuffer::ptr baseBuf = BufferFactory::create(buf);
            while (1)
            {
                if (!_protocol->canProcessed(baseBuf))
                {
                    // 防止一直不可读的脏数据太多
                    if (baseBuf->readableSize() > maxDataSize)
                    {
                        conn->shutdown();
                        std::cerr << "缓冲区内脏数据太多!" << endl;
                        return;
                    }

                    cout << "数据量不足" << endl;
                    break;
                }

                std::cerr << "缓冲区内有数据待处理" << endl;
                BaseMessage::ptr msg;
                bool ret = _protocol->onMessage(baseBuf, msg);
                if (!ret)
                {
                    std::cerr << "缓冲区数据解析失败！" << endl;
                    conn->shutdown();
                    return;
                }

                BaseConnection::ptr baseConn;
                {
                    std::unique_lock<std::mutex> lock(_mtx);
                    auto it = _conns.find(conn);
                    if (it == _conns.end())
                    {
                        conn->shutdown();
                        std::cerr << "非法连接！" << endl;
                        return;
                    }
                    baseConn = it->second;
                }

                // Todo: 回调函数
                if (_messageCb)
                    _messageCb(baseConn, msg);
            }
        }

    private:
        const size_t maxDataSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        muduo::net::EventLoop _loop;
        muduo::net::TcpServer _server;
        std::mutex _mtx;
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::ptr> _conns;
    };

    class ServiceFactory
    {
    public:
        template <typename... Args>
        static BaseService::ptr create(Args... args)
        {
            return std::make_shared<MuduoService>(std::forward<Args>(args)...);
        }
    };

    class MuduoClient : public BaseClient
    {
    public:
        using ptr = std::shared_ptr<MuduoClient>;

        MuduoClient(const string &ip, int port)
            : _protocol(ProtocolFactory::create()), _loop(_loopThread.startLoop()), _countLatch(1), _client(_loop, muduo::net::InetAddress(ip, port), "MuduoClient")
        {
            DLOG("MuduoClient MuduoClient()");
            _client.setConnectionCallback(std::bind(&MuduoClient::onConnection, this, _1));
        }

        virtual void connect() override
        {
            _client.setConnectionCallback(std::bind(&MuduoClient::onConnection, this, _1));
            _client.setMessageCallback(std::bind(&MuduoClient::onMessage, this, _1, _2, _3));
            ILOG("建立连接中...")
            _client.connect();

            // (异步等待连接建立时调用回调函数)防止连接还未建立，客户端就要发送消息
            _countLatch.wait();
            ILOG("连接建立成功...")
        }

        virtual void shutdown() override
        {
            _conn->shutdown();
        }

        virtual bool send(const BaseMessage::ptr msg) override
        {
            if (!connected())
            {
                ELOG("连接已断开...")
                return false;
            }
            _conn->send(msg);
        }

        virtual BaseConnection::ptr connection() override
        {
            return _conn;
        }

        virtual bool connected() override
        {
            return (_conn && _conn->connected());
        }

    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                // 计数 --， 为0时唤醒阻塞
                _countLatch.countDown();
                _conn = ConnectionFactory::create(conn, _protocol);
            }
            else
            {
                ILOG("连接已断开...")
                // 重置共享智能指针
                _conn.reset();
            }
        }

        void onMessage(const muduo::net::TcpConnectionPtr &conn,
                       muduo::net::Buffer *buf,
                       muduo::Timestamp timestamp)
        {
            // ILOG("连接有数据到来，待处理!")
            BaseBuffer::ptr baseBuf = BufferFactory::create(buf);
            while (1)
            {
                // 缓冲区数据不能处理
                if (!_protocol->canProcessed(baseBuf))
                {
                    if (baseBuf->readableSize() > maxDataSize)
                    {
                        ELOG("非法数据太多!")
                        conn->shutdown();
                        return;
                    }
                    break;
                }

                BaseMessage::ptr baseMsg;
                if (!_protocol->onMessage(baseBuf, baseMsg))
                {
                    ELOG("缓冲区数据解析失败！")
                    conn->shutdown();
                    return;
                }

                if (_messageCb)
                    _messageCb(_conn, baseMsg);
            }
        }

    private:
        const size_t maxDataSize = (1 << 16);
        BaseProtocol::ptr _protocol;
        BaseConnection::ptr _conn;
        muduo::net::EventLoopThread _loopThread;
        muduo::net::EventLoop *_loop;
        muduo::net::TcpClient _client;
        muduo::CountDownLatch _countLatch;
    };

    class ClientFactory
    {
    public:
        template <typename... Args>
        static BaseClient::ptr create(Args... args)
        {
            return std::make_shared<MuduoClient>(std::forward<Args>(args)...);
        }
    };

} // rpc